package com.huqing.icu.rentuser.consumer;

import com.huqing.icu.constants.KafkaGroupIdConstants;
import com.huqing.icu.constants.KafkaTopicConstants;
import com.huqing.icu.enums.user.UserNumPanelTypeEnum;
import com.huqing.icu.rentuser.model.User;
import com.huqing.icu.rentuser.service.IUserNumPanelService;
import com.huqing.icu.utils.DateUtils;
import com.huqing.icu.utils.JsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.sql.Timestamp;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/**
 * @Description kafka消费者
 * @Author huqing
 * @Date 2025/9/6 14:42
 **/
@Component
public class KafkaUserConsumer {

    private static final Logger logger = LoggerFactory.getLogger(KafkaUserConsumer.class);

    @Autowired
    private IUserNumPanelService iUserNumPanelService;

    /**
     * 更新 新增用户数、总用户数
     *
     * @param record
     * @param ack
     * @param topic
     */
    @KafkaListener(topics = KafkaTopicConstants.UserPrefix.UPDATE_USER_NUM, groupId = KafkaGroupIdConstants.USER_NUM_GROUP, containerFactory = "kafkaManualAckListenerContainerFactory")
    public void updateUserNum(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

        logger.info("kafka begin consume, topic = {}, message = {}", KafkaTopicConstants.UserPrefix.UPDATE_USER_NUM, record.value());
        Optional message = Optional.ofNullable(record.value());
        if (!message.isPresent()) {
            return;
        }
        Object msg = message.get();
        User user = JsonUtils.getClazz(msg.toString(), User.class);

        //获取用户ID和注册时间
        Long userId = user.getUserId();
        Timestamp createTime = user.getCreateTime();

        //解析注册时间，看用户是哪一天、那一月、哪一年注册的
        String date = DateUtils.timestampToStr(createTime, DateUtils.yyyyMMDDWithBar);
        String month = DateUtils.timestampToStr(createTime, DateUtils.yyyyMMWithBar);
        String year = DateUtils.timestampToStr(createTime, DateUtils.yyyyWithBar);

        List<String> dateList = Arrays.asList(date, month, year);
        List<Integer> typeList = Arrays.asList(UserNumPanelTypeEnum.DAILY.getCode(), UserNumPanelTypeEnum.MONTH.getCode(), UserNumPanelTypeEnum.YEAR.getCode());
        //新增用户数、总用户数 +1
        iUserNumPanelService.incrAddUserNumAndUserNum(dateList, typeList);

        logger.info("kafka consume end, topic = {}, message = {}", KafkaTopicConstants.UserPrefix.UPDATE_USER_NUM, record.value());
        //手动ack
        ack.acknowledge();
    }
}
